package drds.server.util;

import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.zip.Deflater;
import java.util.zip.Inflater;

import com.google.common.collect.Lists;

import drds.server.net.SessionContext$ReadWrite;
import drds.server.net.packet.ByteBufferUtil;
import drds.server.net.packet.BytesHolder;

/**
 * 压缩数据包协议
 * 
 * http://dev.mysql.com/doc/internals/en/compressed-packet-header.html
 * 
 * (包头) 3 Bytes 压缩长度 1 Bytes 压缩序列号 3 Bytes 压缩前的长度
 * 
 * (包体) n Bytes 压缩内容 或 未压缩内容
 * 
 * | --------------------------------------------------------------------------
 * ------------ | | comp-length | seq-id | uncomp-len | Compressed Payload | |
 * ------------------------------------------------
 * ------------------------------------- | | 22 00 00 | 00 | 32 00 00 |
 * compress("\x2e\x00\x00\x00\x03select ...") | |
 * --------------------------------
 * ------------------------------------------------------ |
 * 
 * Q:为什么消息体是 压缩内容 或者未压缩内容? A:这是因为mysql内部有一个约定，如果查询语句payload小于50字节时，
 * 对内容不压缩而保持原貌的方式，而mysql此举是为了减少CPU性能开销
 * 
 */
public class Zip {

	public static final int MINI_LENGTH_TO_COMPRESS = 1024;
	public static final int NO_COMPRESS_PACKET_LENGTH = MINI_LENGTH_TO_COMPRESS + 4;

	/**
	 * 压缩数据包
	 */
	public static ByteBuffer compressPacket(ByteBuffer byteBuffer, SessionContext$ReadWrite sessionContext, ConcurrentLinkedQueue<byte[]> unfinishedBytesQueue) {

		byte[] bytes = getBytes(byteBuffer);
		sessionContext.recycle(byteBuffer);

		bytes = mergeBytes(bytes, unfinishedBytesQueue);
		return compressPacket(bytes, sessionContext, unfinishedBytesQueue);
	}

	/**
	 * 压缩数据包
	 */
	private static ByteBuffer compressPacket(byte[] bytes, SessionContext$ReadWrite sessionContext, ConcurrentLinkedQueue<byte[]> unfinishedBytesQueue) {

		ByteBuffer byteBuffer = sessionContext.allocate();
		// 数据量大的时候,此处是是性能的堵点
		byteBuffer = sessionContext.checkWriteByteBuffer(byteBuffer, bytes.length, false);
		BytesHolder bytesHolder = new BytesHolder(bytes);
		while (bytesHolder.hasRemaining()) {
			// 包体的长度
			int packetLength = 0;
			// 可读的长度
			int readableLength = bytesHolder.length() - bytesHolder.position();
			if (readableLength > 3) {
				packetLength = bytesHolder.readInt3();
				bytesHolder.move(-3);
			}
			// 校验数据包完整性
			if (readableLength < packetLength + 4) {// 4=3+1
				/**
				 * 如果数据不是一个完整的数据包则等待后面的数据到来。则将剩下的数据添加到队列里面。
				 */
				byte[] packetBytes = bytesHolder.readBytes(readableLength);
				if (packetBytes.length != 0) {
					unfinishedBytesQueue.add(packetBytes); // 不完整的包
				}
			} else {

				byte[] packetBytes = bytesHolder.readBytes(packetLength + 4);
				if (packetBytes.length != 0) {

					if (packetBytes.length <= NO_COMPRESS_PACKET_LENGTH) {
						ByteBufferUtil.writeInt3(byteBuffer, packetBytes.length); // 压缩长度
						byteBuffer.put(packetBytes[3]); // 压缩序号
						ByteBufferUtil.writeInt3(byteBuffer, 0); // 压缩前的长度设置为0
						byteBuffer.put(packetBytes); // 包体

					} else {

						byte[] compressedPacketBytes = compress(packetBytes); // 压缩

						ByteBufferUtil.writeInt3(byteBuffer, compressedPacketBytes.length);
						byteBuffer.put(packetBytes[3]);// 压缩序号-该编号不变
						ByteBufferUtil.writeInt3(byteBuffer, packetBytes.length);
						byteBuffer.put(compressedPacketBytes);
					}
				}
			}
		}
		return byteBuffer;
	}

	/**
	 * 解压数据包,同时做分包处理
	 */
	public static List<byte[]> decompressPacket(byte[] bytes, ConcurrentLinkedQueue<byte[]> unfinishedBytesQueue) {

		BytesHolder bytesHolder = new BytesHolder(bytes);

		// 包头
		// -----------------------------------------
		int packetLength = bytesHolder.readInt3(); // 压缩的包长
		@SuppressWarnings("unused")
		byte packetId = bytesHolder.read(); // 压缩的包号
		int oldLength = bytesHolder.readInt3(); // 压缩前的长度

		// 未压缩, 直接返回
		if (packetLength == bytes.length - 4) {
			return Lists.newArrayList(bytes);

			// 压缩不成功的, 直接返回
		} else if (oldLength == 0) {
			byte[] readBytes = bytesHolder.readBytes();
			return splitPacket(readBytes, unfinishedBytesQueue);

			// 解压
		} else {
			byte[] de = decompress(bytes, 7, bytes.length - 7);
			return splitPacket(de, unfinishedBytesQueue);
		}
	}

	/**
	 * 分包处理
	 */
	private static List<byte[]> splitPacket(byte[] bytes, ConcurrentLinkedQueue<byte[]> unfinishedBytesQueue) {

		// 合并
		bytes = mergeBytes(bytes, unfinishedBytesQueue);

		List<byte[]> bytesList = new ArrayList<>();

		BytesHolder bytesHolder = new BytesHolder(bytes);
		while (bytesHolder.hasRemaining()) {

			int readLength = bytesHolder.length() - bytesHolder.position();
			int packetLength = 0;
			if (readLength > 3) {
				packetLength = bytesHolder.readInt3();
				bytesHolder.move(-3);
			}

			if (readLength < packetLength + 4) {
				byte[] packet = bytesHolder.readBytes(readLength);
				if (packet.length != 0) {
					unfinishedBytesQueue.add(packet);
				}

			} else {
				byte[] packet = bytesHolder.readBytes(packetLength + 4);
				if (packet.length != 0) {
					bytesList.add(packet);
				}
			}
		}

		return bytesList;
	}

	/**
	 * 合并未完成的字节
	 */
	private static byte[] mergeBytes(byte[] bytes, ConcurrentLinkedQueue<byte[]> unfinishedBytesQueue) {

		if (!unfinishedBytesQueue.isEmpty()) {

			ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
			try {
				while (!unfinishedBytesQueue.isEmpty()) {
					/**
					 * 弹出字节数组
					 */
					byteArrayOutputStream.write(unfinishedBytesQueue.poll());
				}
				byteArrayOutputStream.write(bytes);
				bytes = byteArrayOutputStream.toByteArray();

			} catch (IOException e) {
				throw new RuntimeException(e);
			} finally {
				try {
					byteArrayOutputStream.close();
				} catch (IOException e) {
				}
			}
		}
		return bytes;
	}

	private static byte[] getBytes(ByteBuffer byteBuffer) {
		byteBuffer.flip();
		byte[] bytes = new byte[byteBuffer.limit()];
		byteBuffer.get(bytes);
		byteBuffer.clear();
		return bytes;
	}

	public static byte[] compress(ByteBuffer byteBuffer) {
		return compress(getBytes(byteBuffer));
	}

	/**
	 * 适用于mysql与客户端交互时zlib 压缩
	 */
	public static byte[] compress(byte[] bytes) {

		byte[] output = null;

		Deflater deflater = new Deflater();
		deflater.setInput(bytes);
		deflater.finish();

		ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(bytes.length);
		byte[] result = new byte[1024];
		try {
			while (!deflater.finished()) {
				int length = deflater.deflate(result);
				byteArrayOutputStream.write(result, 0, length);
			}
			output = byteArrayOutputStream.toByteArray();
		} finally {
			try {
				byteArrayOutputStream.close();
			} catch (Exception e) {
			}
			deflater.end();
		}

		return output;
	}

	/**
	 * 适用于mysql与客户端交互时zlib解压
	 * 
	 * @param bytes
	 *            数据
	 * @param offset
	 *            偏移量
	 * @param length
	 *            长度
	 */
	public static byte[] decompress(byte[] bytes, int offset, int length) {

		byte[] output = null;

		Inflater inflater = new Inflater();
		inflater.reset();
		inflater.setInput(bytes, offset, length);

		ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream(bytes.length);
		try {
			byte[] result = new byte[1024];
			while (!inflater.finished()) {
				int i = inflater.inflate(result);
				byteArrayOutputStream.write(result, 0, i);
			}
			output = byteArrayOutputStream.toByteArray();
		} catch (Exception e) {
			throw new RuntimeException(e);
		} finally {
			try {
				byteArrayOutputStream.close();
			} catch (Exception e) {
			}
			inflater.end();
		}
		return output;
	}

}
